package com.atguigu.day09;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

// DDL的写法
public class Example9 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(env, settings);

        // 定义输入表，WITH是和外部设备进行连接的连接器
        // 将输入转换成动态表
        streamTableEnvironment
                .executeSql("CREATE TABLE clicks (`user` STRING, `url` STRING) " +
                        "WITH (" +
                        "'connector' = 'filesystem'," +
                        "'path' = '/home/zuoyuan/flink0609/src/main/resources/file.csv'," +
                        "'format' = 'csv')");

        // 输出表的定义
        streamTableEnvironment
                .executeSql("CREATE TABLE ResultTable (`user` STRING, `cnt` BIGINT) " +
                        "WITH ('connector' = 'print')");

        // 在动态表上进行查询
        // 查询结果插入到结果表中
        streamTableEnvironment
                .executeSql("INSERT INTO ResultTable " +
                        "SELECT user, COUNT(url) as cnt " +
                        "FROM clicks GROUP BY user");
    }
}
